import copy
import json
import os
import queue
import time
import threading
from loguru import logger
from datetime import datetime, timedelta
from kafka import KafkaConsumer, KafkaProducer
from kafka.errors import NoBrokersAvailable, KafkaError


dir = os.path.dirname(os.path.abspath(__file__))
log_dir = os.path.join(dir, "Log", "logs")
if not os.path.exists(log_dir):
    os.makedirs(log_dir)
# 自定义日志格式
logger.remove()  # 移除默认的日志处理器
logger.add(
    sink=os.path.join(log_dir, "clean_{time:YYYY-MM-DD}.log"),  # 日志文件名带日期
    format="{time:YYYY-MM-DD HH:mm:ss.SSS} | {level} | {function}:{line} - {message}",
    rotation="00:00",           # 每天午夜轮转（与大小轮转二选一）
    retention="360 days",       # 保留最近360天的日志
    encoding="utf-8",           # 避免中文乱码
    level="INFO",               # 文件日志级别
)


class SichuanClean:
    def __init__(self, device_info, config_info):
        self.config = config_info
        self.device_info = device_info
        self.flag = 0
        # self.define = config["Define"]
        self.config_data = config_info["ConfigData"]
        self.current_time = None
        self.time_diff = None
        self.kafka_producer = None
        self.q = queue.Queue(maxsize=1000)
        self.q_ = queue.Queue(maxsize=1000)
        self.kafka_status = 1
        self.duration_time = 0.5
        self.q_dict = {}
        self.clock_status = {}
        self.device_status = {}
        self.device_result_dict = {}
        self.up_device_list = ["A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K"]
        self.down_device_list = ["16", "17", "18", "19", "20", "21", "22", "23", "24", "25", "26", "27", "28", "29", "30"]

    def subscribe_up_data_kafka(self):
        # 将消费者的偏移量设置为最后一条消息的偏移量
        topic = self.config_data["kafka_up_cars_topic"]
        kafka_consumer = None
        while True:
            time.sleep(0.002)
            if kafka_consumer is None:
                try:
                    kafka_consumer = KafkaConsumer(
                        topic,
                        bootstrap_servers=self.config_data["kafka_host"],
                        auto_offset_reset='latest',
                        enable_auto_commit=False
                    )
                    self.kafka_status = 0
                    logger.info(f"上行门架数据监听启动, topic:{topic}")
                except NoBrokersAvailable as e:
                    logger.error(f"kafka连接失败, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    continue

            while True:
                time.sleep(0.0002)
                if self.kafka_status == 1:
                    logger.error("kafka连接断开")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                try:
                    records = kafka_consumer.poll(timeout_ms=1000)
                    if records:
                        for tp, messages in records.items():
                            for message in messages:
                                if message:
                                    message_value = message.value.decode('utf-8')
                                    vehicle_info = json.loads(message_value)
                                    # orgCode = vehicle_info.get('orgCode', "")
                                    # receiveTime = vehicle_info.get('receiveTime', 0)
                                    # sendTime = vehicle_info.get('sendTime', 0)
                                    globalTime = vehicle_info.get('globalTime', 0)
                                    if "targetList" in vehicle_info:
                                        for target in vehicle_info["targetList"]:
                                            id = target.get("id", 0)
                                            carType = target.get("carType", 0)
                                            carColor = target.get("carColor", 0)
                                            staion = target.get("staion", 0)
                                            new_id = f"{id}_{carType}_{carColor}"
                                            vehicle_data = {
                                                "device": staion,
                                                "time": globalTime,
                                                "id": id,
                                                "type": carType,
                                                "color": carColor,
                                                "Id": new_id
                                            }
                                            # self.clock_status[staion]["receive_time"] = receiveTime
                                            # self.clock_status[staion]["send_time"] = sendTime
                                            # self.clock_status[staion]["global_time"] = globalTime
                                            # self.clock_status[staion]["current_time"] = int(time.time() * 1000)
                                            self.q.put(vehicle_data, block=False)
                except queue.Full:
                    self.q.get(block=False)  # 手动丢弃队头数据（1）
                    logger.warning("队列已满，丢弃队头数据")
                except KafkaError as e:
                    logger.error(f"kafka连接断开, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                except Exception as e:
                    logger.opt(exception=e).error(f"解析数据失败, topic:{topic}")
                    pass

    def subscribe_down_data_kafka(self):
        # 将消费者的偏移量设置为最后一条消息的偏移量
        topic = self.config_data["kafka_down_cars_topic"]
        kafka_consumer = None
        while True:
            time.sleep(0.002)
            if kafka_consumer is None:
                try:
                    kafka_consumer = KafkaConsumer(
                        topic,
                        bootstrap_servers=self.config_data["kafka_host"],
                        auto_offset_reset='latest',
                        enable_auto_commit=False
                    )
                    self.kafka_status = 0
                    logger.info(f"下行门架数据监听启动, topic:{topic}")
                except NoBrokersAvailable as e:
                    logger.error(f"kafka连接失败, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    continue

            while True:
                time.sleep(0.0002)
                if self.kafka_status == 1:
                    logger.error("kafka连接断开")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                try:
                    records = kafka_consumer.poll(timeout_ms=1000)
                    if records:
                        for tp, messages in records.items():
                            for message in messages:
                                if message:
                                    message_value = message.value.decode('utf-8')
                                    vehicle_info = json.loads(message_value)
                                    orgCode = vehicle_info.get('orgCode', "")
                                    receiveTime = vehicle_info.get('receiveTime', 0)
                                    sendTime = vehicle_info.get('sendTime', 0)
                                    globalTime = vehicle_info.get('globalTime', 0)
                                    if "targetList" in vehicle_info:
                                        for target in vehicle_info["targetList"]:
                                            id = target.get("id", 0)
                                            carType = target.get("carType", 0)
                                            carColor = target.get("carColor", 0)
                                            staion = target.get("staion", 0)
                                            new_id = f"{id}_{carType}_{carColor}"
                                            vehicle_data = {
                                                "device": staion,
                                                "time": receiveTime,
                                                "id": id,
                                                "type": carType,
                                                "color": carColor,
                                                "Id": new_id
                                            }
                                            self.clock_status[staion]["receive_time"] = receiveTime
                                            self.clock_status[staion]["send_time"] = sendTime
                                            self.clock_status[staion]["global_time"] = globalTime
                                            self.clock_status[staion]["current_time"] = int(time.time() * 1000)
                                            self.q.put(vehicle_data, block=False)
                except queue.Full:
                    self.q.get(block=False)  # 手动丢弃队头数据（1）
                    logger.warning("队列已满，丢弃队头数据")
                except KafkaError as e:
                    logger.error(f"kafka连接断开, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                except Exception as e:
                    logger.opt(exception=e).error(f"解析数据失败, topic:{topic}")
                    pass

    def subscribe_status_kafka(self):
        # 将消费者的偏移量设置为最后一条消息的偏移量
        topic = self.config_data["kafka_status_topic"]
        kafka_consumer = None
        while True:
            time.sleep(0.002)
            if kafka_consumer is None:
                try:
                    kafka_consumer = KafkaConsumer(
                        topic,
                        bootstrap_servers=self.config_data["kafka_host"],
                        auto_offset_reset='latest',
                        enable_auto_commit=False
                    )
                    self.kafka_status = 0
                    logger.info(f"下行门架数据监听启动, topic:{topic}")
                except NoBrokersAvailable as e:
                    logger.error(f"kafka连接失败, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    continue

            while True:
                time.sleep(0.0002)
                if self.kafka_status == 1:
                    logger.error("kafka连接断开")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                try:
                    records = kafka_consumer.poll(timeout_ms=1000)
                    if records:
                        for tp, messages in records.items():
                            for message in messages:
                                if message:
                                    message_value = message.value.decode('utf-8')
                                    status_info = json.loads(message_value)
                                    timeStamp = status_info["timeStamp"]
                                    timestamp = datetime.strptime(timeStamp, "%Y-%m-%d %H:%M:%S:%f")
                                    cameraList = status_info["cameraList"]
                                    status_data = {"timestamp": timestamp,
                                                   "camera_list": cameraList}
                                    self.q_.put(status_data, block=False)
                except queue.Full:
                    self.q_.get(block=False)  # 手动丢弃队头数据（1）
                    logger.warning("队列已满，丢弃队头数据")
                except KafkaError as e:
                    logger.error(f"kafka连接断开, topic:{topic}")
                    kafka_consumer = None
                    time.sleep(10)
                    break
                except Exception as e:
                    logger.opt(exception=e).error(f"解析数据失败, topic:{topic}")
                    pass

    def init(self):
        range_num = len(self.config_data["Gantry"][self.flag])
        for i in range(range_num):
            self.q_dict[i] = []
            self.clock_status[i] = {"receive_time": 0, "send_time": 0, "global_time": 0, "current_time": 0}
            self.device_result_dict[i] = {"total": 0, "positive": 0, "missed": 0, "incorrect": 0, "accuracy": 0, "heartbeat": 1, "clock": 1, "delay": 1}

        # 启动线程实时监听kafka数据
        threading.Thread(target=self.subscribe_up_data_kafka, args=()).start()
        # threading.Thread(target=self.subscribe_down_data_kafka, args=()).start()
        # threading.Thread(target=self.subscribe_status_kafka, args=()).start()

    def hour_data(self, vehicle_list):
        current_time = vehicle_list[-1]["time"]
        k = 0
        for i in range(len(vehicle_list)):
            vehicle_time = vehicle_list[i]["time"]
            if vehicle_time < current_time - 60 * 60 * 1000 * self.duration_time:
                continue
            else:
                k = i
                break
        return k

    def run(self):
        # 循环监听心跳，初始化时间
        last_time = time.time()
        while True:
            time.sleep(0.0002)
            try:
                # kafka 服务器
                if self.kafka_producer is None:
                    logger.info(f"add_message {self.kafka_producer}")
                    try:
                        self.kafka_producer = KafkaProducer(
                            bootstrap_servers=self.config_data["kafka_host"],
                            key_serializer=str.encode,
                            request_timeout_ms=10000,
                            value_serializer=lambda x: json.dumps(x).encode('utf-8')
                        )
                        self.kafka_status = 0
                        logger.info(f"kafka_producer连接成功")
                    except NoBrokersAvailable as e:
                        self.kafka_status = 1
                        logger.error("kafka连接失败")
                        self.kafka_producer = None
                        time.sleep(10)
                        continue
                # 心跳状态上报
                if time.time() - last_time >= 1:
                    m = {
                        "timestamp": str(int(time.time())),
                        "status": 0,
                        "type": 1
                    }
                    try:
                        future = self.kafka_producer.send(self.config_data["kafka_status"], value=m)
                        record_metadata = future.get(timeout=10)  # 正常情况下很快返回
                    except KafkaError as e:
                        self.kafka_status = 1
                        logger.error("kafka连接失败，导致算法心跳上报失败")
                        self.kafka_producer = None
                        time.sleep(10)
                    except Exception as e:
                        logger.error("算法心跳上报失败")
                    # 更新时间
                    last_time = time.time()

                # 从队列中获取最新车流数据
                while not self.q.empty():
                    vehicle_info = self.q.get(block=False)
                    self.q_dict[vehicle_info["gantry_id"]].append(vehicle_info)
                # # 从队列中获取最新设备状态数据
                # while not self.q_.empty():
                #     status_info = self.q_.get(block=False)
                #     for camera in status_info["camera_list"]:
                #         camera_id = camera_device_dict[camera["id"]]
                #         camera_status_list = camera["status"].split(",")
                #         if "2" in camera_status_list:
                #             self.device_result_dict[camera_id]["heartbeat"] = 1
                #         elif camera["status"] == "1":
                #             self.device_result_dict[camera_id]["heartbeat"] = 0
                #         else:
                #             self.device_result_dict[camera_id]["heartbeat"] = camera["status"]
                # 设备状态数据处理，并发送kafka
                if int(time.time()) % 5 == 0:
                    time.sleep(1)
                    # 只保留各基站位置最近一个小时数据进行数据诊断
                    for device_id, vehicle_list in self.q_dict.items():
                        del_index = self.hour_data(vehicle_list)
                        self.q_dict[device_id] = vehicle_list[del_index:]
                        # 获取基站device_id在最近一段时间内的过车总数
                        self.device_result_dict[device_id]["total"] = len(self.q_dict[device_id])

                    # 上行设备精度计算
                    for i in range(len(self.up_device_list) - 1):
                        if i == 0:
                            self.device_result_dict[self.up_device_list[0]]["positive"] = self.device_result_dict[self.up_device_list[0]]["total"]
                            continue
                        positive_id_dict = []
                        up_device_id = self.up_device_list[i]
                        down_device_id = self.up_device_list[i + 1]
                        up_device_data_dict = {vehicle_info["Id"]: vehicle_info for vehicle_info in self.q_dict[up_device_id]}
                        down_device_data_dict = {vehicle_info["Id"]: vehicle_info for vehicle_info in self.q_dict[down_device_id]}
                        for car_id, up_vehicle_info in up_device_data_dict.items():
                            if car_id in down_device_data_dict.keys():
                                positive_id_dict.append(car_id)
                                self.device_result_dict[down_device_id]["positive"] += 1
                        num = self.device_result_dict[down_device_id]["total"] - len(positive_id_dict)
                        self.device_result_dict[down_device_id]["incorrect"] += num
                        # 计算上下游的漏检
                        num_diff = self.device_result_dict[up_device_id]["total"] - self.device_result_dict[down_device_id]["total"]
                        if num_diff > 0:
                            self.device_result_dict[down_device_id]["missed"] += num_diff
                        else:
                            self.device_result_dict[down_device_id]["incorrect"] -= num_diff

                    # 下行设备精度计算
                    for i in range(len(self.down_device_list) - 1):
                        if i == 0:
                            self.device_result_dict[self.down_device_list[0]]["positive"] = self.device_result_dict[self.down_device_list[0]]["total"]
                            continue
                        positive_id_dict = []
                        up_device_id = self.down_device_list[i]
                        down_device_id = self.down_device_list[i + 1]
                        up_device_data_dict = {vehicle_info["Id"]: vehicle_info for vehicle_info in self.q_dict[up_device_id]}
                        down_device_data_dict = {vehicle_info["Id"]: vehicle_info for vehicle_info in self.q_dict[down_device_id]}
                        for car_id, up_vehicle_info in up_device_data_dict.items():
                            if car_id in down_device_data_dict.keys():
                                positive_id_dict.append(car_id)
                                self.device_result_dict[down_device_id]["positive"] += 1
                        num = self.device_result_dict[down_device_id]["total"] - len(positive_id_dict)
                        self.device_result_dict[down_device_id]["incorrect"] += num
                        # 计算上下游的漏检
                        num_diff = self.device_result_dict[up_device_id]["total"] - self.device_result_dict[down_device_id]["total"]
                        if num_diff > 0:
                            self.device_result_dict[down_device_id]["missed"] += num_diff
                        else:
                            self.device_result_dict[down_device_id]["incorrect"] -= num_diff

                    for device_id, data in self.device_result_dict.items():
                        accuracy = data["positive"] / (data["total"] + data["missed"]) * 100
                        self.device_result_dict[device_id]["accuracy"] = round(accuracy, 2)

                    for device_id, data in self.device_result_dict.items():
                        logger.info(f"设备id：{device_id}，设备精度：{data['accuracy']}, 总数：{data['total']}, 正检：{data['positive']}, 误检：{data['incorrect']}, 漏检：{data['missed']}")

                    # # 计算基站状态并发送到kafka
                    # result = {"timestamp": int(time.time()), "devices": []}
                    # for device_id in self.up_device_list:
                    #     clock_status = self.clock_status[device_id]
                    #     delay_time = abs(clock_status["global_time"] - clock_status["send_time"])
                    #     clock_time = abs(clock_status["global_time"] - clock_status["receive_time"])
                    #     # TODO  延迟和时钟偏差大于5s则认为是异常
                    #     self.device_result_dict[device_id]["delay"] = 1 if delay_time > 5 else 0
                    #     self.device_result_dict[device_id]["clock"] = 1 if clock_time > 5 else 0
                    #     # 如果心跳丢失则认为设备参数全部异常
                    #     if self.device_result_dict[device_id]["heartbeat"] == 1:
                    #         self.device_result_dict[device_id]["delay"] = 1
                    #         self.device_result_dict[device_id]["clock"] = 1
                    #         self.device_result_dict[device_id]["accuracy"] = 0
                    #     result_device = {"deviceId": device_id,
                    #                      "heartbeat": self.device_result_dict[device_id]["heartbeat"],
                    #                      "clock": self.device_result_dict[device_id]["delay"],
                    #                      "delay": self.device_result_dict[device_id]["clock"],
                    #                      "precision": self.device_result_dict[device_id]["accuracy"]}
                    #     result["devices"].append(result_device)
                    # self.kafka_producer.send(self.config_data["kafka_device_status"], value=result)

            except Exception as e:
                logger.opt(exception=e).error("数据处理异常")


if __name__ == '__main__':
    # 获取当前文件所在目录
    current_dir = os.path.dirname(__file__)
    print(current_dir)
    # 获取相机设备文件信息
    device_file_path = os.path.join(current_dir, 'device_info.json')
    with open(device_file_path, 'r', encoding='utf-8') as f:
        device_info = json.load(f)
    # print(device_info)
    # 获取配置文件信息
    config_file_path = os.path.join(current_dir, 'config.json')
    with open(config_file_path, 'r', encoding='utf-8') as ff:
        config_info = json.load(ff)
    # print(config_info)
    # 创建对象
    sichuan_clean = SichuanClean(device_info, config_info)
    sichuan_clean.init()
    sichuan_clean.run()